
#include "RtBase.h"
#include "RtReactorBase.h"
#include "RtTimerQueueOrderedList.h"

#ifndef RT_WIN32
    #include <sys/resource.h>
#endif // !RT_WIN32

//////////////////////////////////////////////////////////////////////
// class CRtEventHandlerRepository
//////////////////////////////////////////////////////////////////////

CRtEventHandlerRepository::CRtEventHandlerRepository()
#ifndef RT_WIN32
	: m_pHandlers(NULL)
	, m_nMaxHandler(0)
#endif // !RT_WIN32
{
}

CRtEventHandlerRepository::~CRtEventHandlerRepository()
{
	RT_INFO_TRACE_THIS("CRtEventHandlerRepository::~CRtEventHandlerRepository");
	Close();
}

RtResult CRtEventHandlerRepository::Open()
{
#ifdef RT_WIN32
	RT_ASSERTE_RETURN(m_Handlers.empty(), RT_ERROR_ALREADY_INITIALIZED);

#else
	RT_ASSERTE_RETURN(!m_pHandlers, RT_ERROR_ALREADY_INITIALIZED);


	RtResult rv = SetRlimit(RLIMIT_NOFILE, 8192, m_nMaxHandler);
	if (RT_FAILED(rv))
		return rv;

	m_pHandlers = new CElement[m_nMaxHandler];
	if (!m_pHandlers)
		return RT_ERROR_OUT_OF_MEMORY;
#endif // RT_WIN32

	return RT_OK;
}

RtResult CRtEventHandlerRepository::Close()
{
#ifdef RT_WIN32
	m_Handlers.clear();
#else
	if (m_pHandlers) {
		delete []m_pHandlers;
		m_pHandlers = NULL;
	}
	m_nMaxHandler = 0;
#endif // RT_WIN32
	return RT_OK;
}

#ifndef RT_WIN32

RtResult CRtEventHandlerRepository::SetRlimit(int aResource, int aMaxNum, int &aActualNum)
{
	rlimit rlCur;
	::memset(&rlCur, 0, sizeof(rlCur));
	int nRet = ::getrlimit((__rlimit_resource_t)aResource, &rlCur);
	if (nRet == -1 || rlCur.rlim_cur == RLIM_INFINITY) {
		RT_ERROR_TRACE("CRtEventHandlerRepository::SetRlimit, getrlimit() failed! err=" << errno);
		return RT_ERROR_UNEXPECTED;
	}
	
	aActualNum = aMaxNum;
	if (aActualNum > static_cast<int>(rlCur.rlim_cur)) {
		rlimit rlNew;
		::memset(&rlNew, 0, sizeof(rlNew));
		rlNew.rlim_cur = aActualNum;
		rlNew.rlim_max = aActualNum;
		nRet = ::setrlimit((__rlimit_resource_t)aResource, &rlNew);
		if (nRet == -1) {
			if (errno == EPERM) {
				RT_WARNING_TRACE("CRtEventHandlerRepository::SetRlimit, setrlimit() failed. "
					"you should use superuser to setrlimit(RLIMIT_NOFILE)!");
				aActualNum = rlCur.rlim_cur;
			}
			else {
				RT_WARNING_TRACE("CRtEventHandlerRepository::SetRlimit, setrlimit() failed! err=" << errno);
				return RT_ERROR_UNEXPECTED;
			}
		}
	}
	else
		aActualNum = rlCur.rlim_cur;
	
	return RT_OK;
}
#endif // !RT_WIN32

inline void FdSet_s(fd_set &aFsRead, fd_set &aFsWrite, fd_set &aFsException, 
					CRtEventHandlerRepository::CElement &aEleGet, 
					int &aMaxFd)
{
	int nSocket = (int)aEleGet.m_pEh->GetHandle();
	if (nSocket > aMaxFd)
		aMaxFd = nSocket;

	// READ, ACCEPT, and CONNECT flag will place the handle in the read set.
	if (RT_BIT_ENABLED(aEleGet.m_Mask, IRtEventHandler::READ_MASK) || 
		RT_BIT_ENABLED(aEleGet.m_Mask, IRtEventHandler::ACCEPT_MASK) || 
		RT_BIT_ENABLED(aEleGet.m_Mask, IRtEventHandler::CONNECT_MASK))
	{
		FD_SET(nSocket, &aFsRead);
	}
	// WRITE and CONNECT flag will place the handle in the write set.
	if (RT_BIT_ENABLED(aEleGet.m_Mask, IRtEventHandler::WRITE_MASK) || 
		RT_BIT_ENABLED(aEleGet.m_Mask, IRtEventHandler::CONNECT_MASK))
	{
		FD_SET(nSocket, &aFsWrite);
	}
#ifdef RT_WIN32
	if (RT_BIT_ENABLED(aEleGet.m_Mask, IRtEventHandler::CONNECT_MASK))
	{
		FD_SET(nSocket, &aFsException);
	}
#endif // RT_WIN32
}

int CRtEventHandlerRepository::
FillFdSets(fd_set &aFsRead, fd_set &aFsWrite, fd_set &aFsException)
{
	int nMaxFd = -1;
#ifdef RT_WIN32
	HandlersType::iterator iter = m_Handlers.begin();
	for ( ; iter != m_Handlers.end(); ++iter) {
		CElement &eleGet = (*iter).second;
		FdSet_s(aFsRead, aFsWrite, aFsException, eleGet, nMaxFd);
	}
#else
	for (int i = 0; i < m_nMaxHandler; i++) {
		CElement &eleGet = m_pHandlers[i];
		if (!eleGet.IsCleared())
			FdSet_s(aFsRead, aFsWrite, aFsException, eleGet, nMaxFd);
	}
#endif // RT_WIN32
	return nMaxFd;
}


//////////////////////////////////////////////////////////////////////
// class CRtReactorBase
//////////////////////////////////////////////////////////////////////

CRtReactorBase::CRtReactorBase(PROPERTY aProperty)
	: IRtReactor(aProperty)
	, m_pTimerQueue(NULL)
#ifdef RT_WIN32
	, m_bExclusive(FALSE)
	, m_bOverlapped(FALSE)
#endif
{
}

CRtReactorBase::~CRtReactorBase()
{
	// needn't do Close() because the inherited class will do it
//	Close();
}

RtResult CRtReactorBase::Open()
{
	m_Est.Reset2CurrentThreadId();
	CRtEventQueueUsingMutex::Reset2CurrentThreadId();
	CRtStopFlag::m_Est.Reset2CurrentThreadId();
	
	// check whether inheried class instanced the timer queue.
	if (!m_pTimerQueue) {
		m_pTimerQueue = new CRtTimerQueueOrderedList(NULL);
		if (!m_pTimerQueue)
			return RT_ERROR_OUT_OF_MEMORY;
	}

	RtResult rv = m_EhRepository.Open();
	if (RT_FAILED(rv))
		return rv;

	return rv;
}

RtResult CRtReactorBase::
RegisterHandler(IRtEventHandler *aEh, IRtEventHandler::MASK aMask)
{
	// FIXME TODO: Register handler after OnClose!
	
	m_Est.EnsureSingleThread();
	RtResult rv;
	RT_ASSERTE_RETURN(aEh, RT_ERROR_INVALID_ARG);
	
	IRtEventHandler::MASK maskNew = aMask & IRtEventHandler::ALL_EVENTS_MASK;
	if (maskNew == IRtEventHandler::NULL_MASK) {
		RT_WARNING_TRACE("CRtReactorBase::RegisterHandler, NULL_MASK. aMask=" << aMask);
		return RT_ERROR_INVALID_ARG;
	}

	CRtEventHandlerRepository::CElement eleFind;
	RT_HANDLE fdNew = aEh->GetHandle();
	rv = m_EhRepository.Find(fdNew, eleFind);
	if (maskNew == eleFind.m_Mask && aEh == eleFind.m_pEh) {
//		RT_WARNING_TRACE("CRtReactorBase::RegisterHandler, mask is equal."
//			" aEh=" << aEh <<
//			" aMask=" << aMask <<
//			" fdNew=" << fdNew <<
//			" rv=" << rv);
		return RT_OK;
	}

	if (eleFind.IsCleared()) {
		rv = OnHandleRegister(fdNew, maskNew, aEh);

		// needn't remove handle when OnHandleRegister() failed
		// because the handle didn't be inserted at all
		if (RT_FAILED(rv))
			return rv;
	}

	CRtEventHandlerRepository::CElement eleNew(aEh, maskNew);
	rv = m_EhRepository.Bind(fdNew, eleNew);
	return rv;
}

RtResult CRtReactorBase::
RemoveHandler(IRtEventHandler *aEh, IRtEventHandler::MASK aMask)
{
	m_Est.EnsureSingleThread();
	RtResult rv;
	RT_ASSERTE_RETURN(aEh, RT_ERROR_INVALID_ARG);
	
	IRtEventHandler::MASK maskNew = aMask & IRtEventHandler::ALL_EVENTS_MASK;
	if (maskNew == IRtEventHandler::NULL_MASK) {
		RT_WARNING_TRACE("CRtReactorBase::RemoveHandler, NULL_MASK. aMask=" << aMask);
		return RT_ERROR_INVALID_ARG;
	}

	CRtEventHandlerRepository::CElement eleFind;
	RT_HANDLE fdNew = aEh->GetHandle();
	rv = m_EhRepository.Find(fdNew, eleFind);
	if (RT_FAILED(rv)) {
//		RT_WARNING_TRACE("CRtReactorBase::RemoveHandler, handle not registed."
//			" aEh=" << aEh <<
//			" aMask=" << aMask <<
//			" fdNew=" << fdNew <<
//			" rv=" << rv);
		return rv;
	}

	rv = RemoveHandleWithoutFinding_i(fdNew, eleFind, maskNew);
	return rv;
}

RtResult CRtReactorBase::Close()
{
	RT_INFO_TRACE_THIS("CRtReactorBase::Close m_pTimerQueue="<<m_pTimerQueue);
	if (m_pTimerQueue) {
		// I am sorry to comment it because PostMessage(WM_QUIT) will fail if
		// in atexit() route.
//		m_Est.EnsureSingleThread();
		delete m_pTimerQueue;
		m_pTimerQueue = NULL;
	}
	m_EhRepository.Close();
	RT_INFO_TRACE_THIS("CRtReactorBase::Close after m_EhRepository.Close");
	CRtEventQueueBase::DestoryPendingEvents();
	RT_INFO_TRACE_THIS("CRtReactorBase::Close after DestoryPendingEvents");
	return RT_OK;
}

RtResult CRtReactorBase::
ScheduleTimer(IRtTimerHandler *aTh, LPVOID aArg, 
			  const CRtTimeValue &aInterval, DWORD aCount)
{
	m_Est.EnsureSingleThread();
	if (!m_pTimerQueue) {
		RT_WARNING_TRACE("CRtReactorBase::ScheduleTimer, m_pTimerQueue not inited or closed.");
		return RT_ERROR_NOT_INITIALIZED;
	}

	return m_pTimerQueue->ScheduleTimer(aTh, aArg, aInterval, aCount);
}

RtResult CRtReactorBase::CancelTimer(IRtTimerHandler *aTh)
{
	m_Est.EnsureSingleThread();
	if (!m_pTimerQueue) {
		RT_WARNING_TRACE("CRtReactorBase::CancelTimer, m_pTimerQueue not inited or closed.");
		return RT_ERROR_NOT_INITIALIZED;
	}

	return m_pTimerQueue->CancelTimer(aTh);
}

RtResult CRtReactorBase::
ProcessHandleEvent(RT_HANDLE aFd, IRtEventHandler::MASK aMask, 
				   RtResult aReason, BOOL aIsNotify, BOOL aDropConnect)
{
	m_Est.EnsureSingleThread();
	if (aFd == RT_INVALID_HANDLE) {
		RT_ASSERTE(aMask == IRtEventHandler::EVENTQUEUE_MASK);

#ifdef RT_WIN32
		if (m_bExclusive) {
			RT_ERROR_TRACE("CRtReactorBase::ProcessHandleEvent() should not be overlapped here!");
			m_bOverlapped = TRUE;
			return RT_OK;
		}
		m_bExclusive = TRUE;
#endif

		DWORD dwRemainSize = 0;
		CRtEventQueueBase::EventsType listEvents;
		RtResult rv = CRtEventQueueUsingMutex::PopPendingEventsWithoutWait(
			listEvents, CRtEventQueueBase::MAX_GET_ONCE, &dwRemainSize);
		if (RT_SUCCEEDED(rv))
			rv = CRtEventQueueBase::ProcessEvents(listEvents);

#ifdef RT_WIN32
		if (dwRemainSize || m_bOverlapped) 
			NotifyHandler(NULL, IRtEventHandler::EVENTQUEUE_MASK);

		m_bExclusive = FALSE;
		m_bOverlapped = FALSE;
#else
		if (dwRemainSize)
			NotifyHandler(NULL, IRtEventHandler::EVENTQUEUE_MASK);
#endif

		return rv;
	}

#ifndef RT_DISABLE_EVENT_REPORT
	CRtTimeValue tvCur = CRtTimeValue::GetTimeOfDay();
#endif // !RT_DISABLE_EVENT_REPORT

	CRtEventHandlerRepository::CElement eleFind;
	RtResult rv = m_EhRepository.Find(aFd, eleFind);
	if (RT_FAILED(rv)) {
		if (!aDropConnect) {
			RT_WARNING_TRACE("CRtReactorBase::ProcessHandleEvent, handle not registed."
				" aFd=" << aFd <<
				" aMask=" << aMask <<
				" aReason=" << aReason <<
				" rv=" << rv);
		}
		return rv;
	}

	if (RT_BIT_DISABLED(aMask, IRtEventHandler::CLOSE_MASK)) {
		IRtEventHandler::MASK maskActual = eleFind.m_Mask & aMask;
		// needn't check the registered mask if it is notify.
		if (!maskActual && !aIsNotify) {
			RT_WARNING_TRACE("CRtReactorBase::ProcessHandleEvent, mask not registed."
				" aFd=" << aFd <<
				" aMask=" << aMask <<
				" m_Mask=" << eleFind.m_Mask <<
				" aReason=" << aReason);
			return RT_OK;
		}
		
		int nOnCall = 0;
		if (aDropConnect && maskActual & IRtEventHandler::CONNECT_MASK) {
			RT_WARNING_TRACE("CRtReactorBase::ProcessHandleEvent, drop connect."
				" aFd=" << aFd <<
				" aMask=" << aMask <<
				" m_Mask=" << eleFind.m_Mask);
			nOnCall = -1;
		}
		else {
			if (maskActual & IRtEventHandler::ACCEPT_MASK
				|| maskActual & IRtEventHandler::READ_MASK)
			{
				nOnCall = eleFind.m_pEh->OnInput(aFd);
			}
			if ((nOnCall == 0 || nOnCall == -2) && 
				(maskActual & IRtEventHandler::CONNECT_MASK
				|| maskActual & IRtEventHandler::WRITE_MASK))
			{
				nOnCall = eleFind.m_pEh->OnOutput(aFd);
			}
		}

		if (nOnCall == 0) {
			rv = RT_OK;
		}
		else if (nOnCall == -2) {
			rv = RT_ERROR_WOULD_BLOCK;
		} 
		else {
			// maybe the handle is reregiested or removed when doing callbacks. 
			// so we have to refind it.
			CRtEventHandlerRepository::CElement eleFindAgain;
			rv = m_EhRepository.Find(aFd, eleFindAgain);
			if (RT_FAILED(rv) || eleFind.m_pEh != eleFindAgain.m_pEh) {
				//RT_ERROR_TRACE("CRtReactorBase::ProcessHandleEvent,"
				//	" callback shouldn't return fail after the fd is reregiested or removed!"
				//	" aFd=" << aFd << 
				//	" EHold=" << eleFind.m_pEh << 
				//	" EHnew=" << eleFindAgain.m_pEh << 
				//	" find=" << rv);
				//////////////////////////////////////////////////////////////////////////
				/// Webb: It is possible for RUDP to receive a bad handshake pdu from peer side and 
				///       then RemoveHandler() from the reactor. 
				//RT_ASSERTE(FALSE);
			}
			else {
				rv = RemoveHandleWithoutFinding_i(aFd, eleFindAgain, 
					IRtEventHandler::ALL_EVENTS_MASK | IRtEventHandler::SHOULD_CALL);
			}
			rv = RT_ERROR_FAILURE;
		}
	}
	else {
//		RT_INFO_TRACE("CRtReactorBase::ProcessHandleEvent, handle is closed."
//			" aFd=" << aFd <<
//			" aMask=" << aMask <<
//			" aReason=" << aReason);

		rv = RemoveHandleWithoutFinding_i(aFd, eleFind, 
			IRtEventHandler::ALL_EVENTS_MASK | IRtEventHandler::SHOULD_CALL);
		rv = RT_ERROR_FAILURE;
	}

#ifndef RT_DISABLE_EVENT_REPORT
	CRtTimeValue tvSub = CRtTimeValue::GetTimeOfDay() - tvCur;
	if (tvSub > CRtEventQueueBase::s_tvReportInterval) {
		RT_WARNING_TRACE_THIS("CRtReactorBase::ProcessHandleEvent, report,"
			" sec=" << tvSub.GetSec() << 
			" usec=" << tvSub.GetUsec() <<
			" aFd=" << aFd <<
			" aMask=" << aMask <<
			" maskFind=" << eleFind.m_Mask << 
			" ehFind=" << eleFind.m_pEh << 
			" aReason=" << aReason);
	}
#endif // !RT_DISABLE_EVENT_REPORT
	return rv;
}

RtResult CRtReactorBase::ProcessTimerTick()
{
#ifndef RT_DISABLE_EVENT_REPORT
	CRtTimeValue tvCur = CRtTimeValue::GetTimeOfDay();
#endif // !RT_DISABLE_EVENT_REPORT

	m_Est.EnsureSingleThread();
	RT_ASSERTE_RETURN(m_pTimerQueue, RT_ERROR_NOT_INITIALIZED);
	if (m_pTimerQueue) 
		m_pTimerQueue->CheckExpire();

#ifndef RT_DISABLE_EVENT_REPORT
	CRtTimeValue tvSub = CRtTimeValue::GetTimeOfDay() - tvCur;
	if (tvSub > CRtEventQueueBase::s_tvReportInterval) {
		RT_WARNING_TRACE_THIS("CRtReactorBase::ProcessTimerTick, report,"
			" sec=" << tvSub.GetSec() << 
			" usec=" << tvSub.GetUsec());
	}
#endif // !RT_DISABLE_EVENT_REPORT
	return RT_OK;
}

RtResult CRtReactorBase::
RemoveHandleWithoutFinding_i(RT_HANDLE aFd, 
						   const CRtEventHandlerRepository::CElement &aHe, 
						   IRtEventHandler::MASK aMask)
{
	IRtEventHandler::MASK maskNew = aMask & IRtEventHandler::ALL_EVENTS_MASK;
	IRtEventHandler::MASK maskEh = aHe.m_Mask;
	IRtEventHandler::MASK maskSelect = (maskEh & maskNew) ^ maskEh;
	if (maskSelect == maskEh) {
		RT_WARNING_TRACE("CRtReactorBase::RemoveHandleWithoutFinding_i, mask is equal. aMask=" << aMask);
		return RT_OK;
	}

	if (maskSelect == IRtEventHandler::NULL_MASK) {
		RtResult rv = m_EhRepository.UnBind(aFd);
		if (RT_FAILED(rv)) {
			RT_WARNING_TRACE("CRtReactorBase::RemoveHandleWithoutFinding_i, UnBind() failed!"
				" aFd=" << aFd <<
				" aMask=" << aMask <<
				" rv=" << rv);
		}
		OnHandleRemoved(aFd);
		if (aMask & IRtEventHandler::SHOULD_CALL) {
			aHe.m_pEh->OnClose(aFd, maskEh);
		}
		return RT_OK;
	}
	else {
		CRtEventHandlerRepository::CElement eleBind = aHe;
		eleBind.m_Mask = maskSelect;
		RtResult rvBind = m_EhRepository.Bind(aFd, eleBind);
		RT_ASSERTE(rvBind == RT_ERROR_FOUND);
		return rvBind;
	}
}

// this function can be invoked in the different thread.
RtResult CRtReactorBase::SendEvent(IRtEvent *aEvent)
{
	return CRtEventQueueUsingMutex::SendEvent(aEvent);
}

// this function can be invoked in the different thread.
RtResult CRtReactorBase::PostEvent(IRtEvent* aEvent, EPriority aPri)
{
	DWORD dwOldSize = 0;
	RtResult rv = CRtEventQueueUsingMutex::
		PostEventWithOldSize(aEvent, aPri, &dwOldSize);
	if (RT_SUCCEEDED(rv) && dwOldSize == 0)
		return NotifyHandler(NULL, IRtEventHandler::EVENTQUEUE_MASK);
	return rv;
}

// this function can be invoked in the different thread.
DWORD CRtReactorBase::GetPendingEventsCount()
{
	return CRtEventQueueUsingMutex::GetPendingEventsCount();
}
